package cn.doitedu.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class FlinkUtils {

    public static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    public static DataStream<String> createKafkaStream(ParameterTool parameterTool) {

        //设置全局参数，就是将客户端的参数，先发送给jobManager，然后再广播到所有的TaskManager
        env.getConfig().setGlobalJobParameters(parameterTool);
        String topicStr = parameterTool.getRequired("kafka.input.topics");
        List<String> topicList = Arrays.asList(topicStr.split(","));

        Properties properties = new Properties();
        //为了容错，要开启checkpoint
        //env.enableCheckpointing(parameterTool.getLong("checkpoint.interval", 60000));
        //将job cancel后，保留外部存储的checkpoint数据（为了以后重新提交job恢复数据）
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        properties.setProperty("bootstrap.servers", parameterTool.getRequired("bootstrap.servers"));
        properties.setProperty("auto.offset.reset", parameterTool.get("auto.offset.reset", "earliest"));


        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>(
                topicList,
                new SimpleStringSchema(),
                properties
        );

        return env.addSource(kafkaConsumer);

    }
}
